package dacp.etl.kafka.steam;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.HashSet;
import java.util.Iterator;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Objects;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;

public class Chain {
	private static final Logger log = LoggerFactory.getLogger(Chain.class);

	Pos root;

	public static Integer[] s1;
	public static Integer[] s2;
	public static Integer[] s3;
	public static Integer[] s4;

	public static void setS1(Integer[] s1){
		Chain.s1 = s1;
	}

	public static void setS2(Integer[] s2){
		Chain.s2 = s2;
	}
	public static void setS3(Integer[] s3){
		Chain.s3 = s3;
	}
	public static void setS4(Integer[] s4){
		Chain.s4 = s4;
	}


	@Override
	public String toString() {
		return Objects.toStringHelper(this).add("root", root).toString();
	}

	public ArrayList<String> process(String ori) {
		Pos cur = root;
		Iterator<String> iterator = S1MMEIN.spliter.split(ori).iterator();
		int i = 0;
		ArrayList<SortStruct> sortStructlist = Lists.newArrayList();

		while (iterator.hasNext()) {
			String next = iterator.next();
			if (cur.inx == i) {
				//list.add(cur.trans(next));
				sortStructlist.add(new SortStruct(cur.beforeSortInx, cur.trans(next)));
				cur = cur.next;
				if (cur == null) {
					break;
				}
			}
			i++;
		}
		return sort(sortStructlist);
		//ArrayList<String> list = sort(sortStructlist);
		//return S1MMEIN.joiner.join(list);
	}

	private class SortStruct {
		int inx;
		String value;

		public SortStruct(int inx, String value) {
			this.inx = inx;
			this.value = value;
		}
	}

	private ArrayList<String> sort(ArrayList<SortStruct> sortStructlist){
		ArrayList<String> list = Lists.newArrayList();
		SortStruct[] array = sortStructlist.toArray(new SortStruct[] {});
		Arrays.sort(array, new Comparator<SortStruct>() {
			@Override
			public int compare(SortStruct o1, SortStruct o2) {
				return o1.inx - o2.inx;
			}
		});
		for (SortStruct a : array) {
			list.add(a.value);
		}
		return list;
	}
	
	public Object[] processPair(String ori) {
		Pos cur = root;
		ArrayList<String> listKey = Lists.newArrayList();

		ArrayList<SortStruct> sortStructlist = Lists.newArrayList();
		Iterator<String> iterator = S1MMEIN.spliter.split(ori).iterator();
		int i = 0;
		while (iterator.hasNext()) {
			String next = iterator.next();
			if (cur.inx == i) {
				String trans = cur.trans(next);
				// list.add(trans);
				sortStructlist.add(new SortStruct(cur.beforeSortInx, trans));
				if (S1MMEIN.keyColsSet.contains(i)) {
					listKey.add(trans);
				}
				if ((cur = cur.next) == null) {
					break;
				}
			}
			i++;
		}
		ArrayList<String> list = sort(sortStructlist);

		return new Object[]{S1MMEIN.keyJoiner.join(listKey), list} ;
		//return new String[] { S1MMEIN.keyJoiner.join(listKey), S1MMEIN.joiner.join(list) };
	}

	
	static class Pari{
		int inx;
		int orderInx;
		public Pari(int inx, int orderInx){
			this.inx= inx;
			this.orderInx = orderInx;
		}
	}
	
	public static Chain build() {
		Chain chain = new Chain();
		log.info("build chain split: {}, trans: {}, dels: {}", s1, s2, s3);

		Integer[] copyOfs1 = Arrays.copyOf(s1, s1.length);
		
		Pari[] ps = new Pari[copyOfs1.length];
		for(int i=0; i<copyOfs1.length; i++){
			ps[i] = new Pari(copyOfs1[i], i);
		} 
		
		Arrays.sort(ps, new Comparator<Pari>() {
			@Override
			public int compare(Pari o1, Pari o2) {
				return o1.inx - o2.inx;
			}
		}); //5,7,9

		boolean time = s2 != null && s2.length > 0;
		HashSet<Integer> h2 = Sets.newHashSet(Arrays.asList(s2));
		TimeProcessor tp = new TimeProcessor(S1MMEIN.DATETIME_FORMAT);

		boolean del = s3 != null && s3.length > 0;
		HashSet<Integer> h3 = Sets.newHashSet(Arrays.asList(s3));
		DelProcessor dp = new DelProcessor(S1MMEIN.DELETE_CHAR);

		boolean trim = s4 != null && s4.length > 0;
		HashSet<Integer> h4 = Sets.newHashSet(Arrays.asList(s4));
		TrimProcessor trimProcessor = new TrimProcessor();

		int i = 0;
		Pos cur = null;
		for (Pari p : ps) {
			Pos pos = new Pos(p.inx, p.orderInx);

			if (time && h2.contains(p.inx)) {
				pos.addProcessor(tp);
			}
			if (del && h3.contains(p.inx)) {
				pos.addProcessor(dp);
			}
			if (trim && h4.contains(p.inx)) {
				pos.addProcessor(trimProcessor);
			}

			if (i == 0) {
				chain.root = pos;
			} else {
				cur.next = pos;
			}
			cur = pos;
			i++;
		}
		log.info("build.chain : {}", chain.toString());
		return chain;
	}

}
